package com.jml.disruptort_more_more;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;

public class Main {
//多个生产者，和多个消费者：
    public static void main(String[] args) throws Exception {

        //创建ringBuffer,业务比较简单就用Ringbuffer，否则就用disrupter
        RingBuffer<Order> ringBuffer =
                //第一个参数表示多个生产者
                //第二个参数相当于factory
                //第三个参数就是RingBuffer的大小
                //第四个参数是策略
                RingBuffer.create( ProducerType.MULTI,
                        new EventFactory<Order>() {
                            @Override
                            public Order newInstance() {
                                return new Order();
                            }
                        },
                        1024 * 1024,
                        new YieldingWaitStrategy());

        SequenceBarrier barriers = ringBuffer.newBarrier();

        Consumer[] consumers = new Consumer[3];
        for (int i = 0; i < consumers.length; i++) {
            consumers[i] = new Consumer("c" + i);
        }
        //多个消费者放入WookPool,生成Sequence
        WorkerPool<Order> workerPool =
                new WorkerPool<Order>(ringBuffer,
                        barriers,
                        //异常处理
                        new ExceptionHandler() {
                            @Override
                            public void handleEventException(Throwable ex, long sequence, Object event) {
                            }
                            @Override
                            public void handleOnStartException(Throwable ex) {
                            }
                            @Override
                            public void handleOnShutdownException(Throwable ex) {
                            }
                        },
                        consumers);
        //把消费者消费进度的下标发送给生产者，让生产者知道消费进度
        //把sequence放入ringBuffer
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
        //创建固定数量的线程池
        workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));

        final CountDownLatch latch = new CountDownLatch(1);
        for (int i = 0; i < 100; i++) {
            //100个生产者，创建100个线程
            final Producer p = new Producer(ringBuffer);
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        latch.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    for (int j = 0; j < 100; j++) {
                        //每个线程提交100个任务
                        p.onData(UUID.randomUUID().toString());
                    }
                }
            }).start();
        }
        Thread.sleep(2000);
        System.out.println("---------------开始生产-----------------");
        //同时生产
        latch.countDown();
        Thread.sleep(5000);
        System.out.println("总数:" + consumers[0].getCount());
    }
}
